package com.afdor.rws.event.channel.jedis.sharded;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.event.Event;
import com.afdor.rws.event.channel.client.consumer.AbstractEventConsumer;
import com.afdor.rws.event.channel.client.consumer.EventConsumer;
import com.afdor.rws.event.channel.client.consumer.handler.EventRetryStrategy;
import com.afdor.rws.event.channel.jedis.JedisUtils;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPool;

/**
 * 基于redis的事件消费者实现
 *
 * @author 悭梵
 * @date Created in 2018-08-17 10:24
 */
@Slf4j
@ToString
@EqualsAndHashCode
@Extension(value = "sharded_jedis", order = Extension.NORMAL_ORDER - 100, scope = Extension.Scope.PROTOTYPE, interfaceClass = EventConsumer.class)
public class ShardedJedisEventConsumer extends AbstractEventConsumer {

    @Getter
    @Setter
    private ShardedJedisPool shardedJedisPool;

    private String queueName;

    public ShardedJedisEventConsumer(@NonNull ShardedJedisPool shardedJedisPool) {
        this.shardedJedisPool = shardedJedisPool;
    }


    @Override
    protected void doStart() {
        super.doStart();
        buildQueueName(getEventSourceName());
    }

    @Override
    protected boolean consume() {
        ShardedJedis jedis = shardedJedisPool.getResource();
        EventRetryStrategy retryStrategy = getEventRetryStrategy();
        String result = null;
        if (retryStrategy != null && retryStrategy instanceof JedisEventRetryStrategy) {
            // 重试队列中获取
            result = ((JedisEventRetryStrategy) retryStrategy).getRetryResult(jedis);
        }
        if (StringUtils.isBlank(result)) {
            result = jedis.lpop(this.queueName);
        }
        if (StringUtils.isBlank(result)) {
            return Boolean.FALSE;
        }
        Event event = getSerialization().unserialize(result);
        if (event == null) {
            return Boolean.FALSE;
        }
        try {
            if (log.isInfoEnabled()) {
                log.info("<===== 收到消息：" + event);
            }
            getEventHandler().handler(event);
        } catch (Exception e) {
            if (retryStrategy != null) {
                if (retryStrategy instanceof JedisEventRetryStrategy) {
                    ((JedisEventRetryStrategy) retryStrategy).retry(event, jedis);
                } else {
                    retryStrategy.retry(event);
                }
            }
            log.error("事件处理异常", e);
        } finally {
            jedis.close();
        }
        return Boolean.TRUE;
    }

    /**
     * 获取队列名称
     *
     * @param eventSourceName
     * @return
     */
    private String buildQueueName(String eventSourceName) {
        if (queueName == null) {
            queueName = JedisUtils.getQueueName(eventSourceName);
        }
        return queueName;
    }

}
